package com.zyx.pulsardemo.basic.consumers;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;

import com.zyx.pulsardemo.basic.base.PulsarDemoBase;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;

/**
 * @author Yaxi.Zhang
 * @since 2021/9/30 15:16
 * reference: https://github.com/david-streamlio/pulsar-in-action
 */
public abstract class PulsarConsumerDemoBase extends PulsarDemoBase {

    protected void startConsumer() {
        final Consumer<byte[]> consumer = getConsumer();
        startConsumer(consumer);
    }

    protected void startConsumer(final Consumer<byte[]> consumer) {
        try {
            consumer.seek(MessageId.earliest);
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
        Runnable run = () -> {
            while (true) {
                Message<byte[]> msg = null;
                // Wait for a message
                try {
                    // 处理消息
                    /*msg = getConsumer().receive();*/ // 示例代码有问题
                    msg = consumer.receive();
                    System.out.printf("Message received: %s \n", new String(msg.getData()));
                    // 确认消息, 以便Broker可以删除该消息
                    /*getConsumer().acknowledge(msg);*/ // 示例代码有问题
                    consumer.acknowledgeAsync(msg);
                    Thread.sleep(10);
                } catch (Exception e) {
                    System.err.printf("Unable to consume message: %s \n", e.getMessage());
                    // 确认消息, 以便Broker可以删除该消息
                    consumer.negativeAcknowledge(msg);
                }
            }
        };
        new Thread(run).start();
    }

    protected abstract Consumer<byte[]> getConsumer();
}
